package org.example;

import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.bson.Document;

import java.util.function.Consumer;


public class StreamingJob {
    public static void main(String[] args) throws Exception {
        // see https://docs.mongodb.com/manual/changeStreams/#open-a-change-stream

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // make sure not pass an iterator to stream source, cause the iterator (ClusterId) cannot be serialized.
        DataStream<Tuple2<String, Integer>> dataStream = env
            .addSource(new SourceFunction<Document>() {
                @Override
                public void run(SourceContext<Document> ctx) throws Exception {
                    new MongoClient(new MongoClientURI("mongodb://root:123456@mongo:27017"))
                        .getDatabase("demo")
                        .getCollection("change")
                        .watch()
                        .forEach((Consumer<? super ChangeStreamDocument<Document>>) c -> ctx.collect(c.getFullDocument()));
                }

                @Override
                public void cancel() {

                }
            })
            .flatMap(new Splitter())
            .keyBy(0)
            .timeWindow(Time.seconds(5))
            .sum(1);

        dataStream.print();

        env.execute("Window WordCount");
    }

    public static class Splitter implements FlatMapFunction<Document, Tuple2<String, Integer>> {
        @Override
        public void flatMap(Document document, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word : document.getString("data").split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }

}
